package top.wangjiu.consume;


import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import top.wangjiu.reactive.rpc.config.ClientConfiguration;
import top.wangjiu.reactive.rpc.config.DefaultServiceConfig;
import top.wangjiu.reactive.rpc.registration.LocalConfigServerAddrList;
import top.wangjiu.reactive.rpc.util.RpcUtil;
import top.wangjiu.service.interfaces.OneService;
import top.wangjiu.service.interfaces.TwoService;

import java.util.concurrent.atomic.AtomicInteger;


public class ConsumeApp {

    public static void main(String[] args) {

        try {
            ClientConfiguration configuration = new ClientConfiguration();
            LocalConfigServerAddrList addrList = new LocalConfigServerAddrList();
            addrList.addService(OneService.class,0,"localhost",9000);
            addrList.addService(TwoService.class,0,"localhost",9000);
            configuration.setServerAddrList(addrList);
            OneService oneService = RpcUtil.getRpcInstance(OneService.class,new DefaultServiceConfig(),configuration);
            TwoService twoService = RpcUtil.getRpcInstance(TwoService.class,new DefaultServiceConfig(),configuration);

            //响应式，Flux,测试回压功能，
            Thread[] threads = new Thread[5];
            for (int i=0;i<5;i++){
                threads[i] = new Thread(()->{
                    int start = 0;
                    int end = 10;
                    //模拟每次请求3个资源
                    int requestPageSize = 3;
                    AtomicInteger count = new AtomicInteger(0);
                    //向生产者获取从start到end的数字列表，测试回压
                    Flux<Integer> list = oneService.getList(start, end);
                    list.subscribe(new Subscriber<Integer>() {
                        Subscription s = null;
                        @Override
                        public void onSubscribe(Subscription s) {
                            this.s = s;
                            //与生产者建立了通道，开启请求第一批数据
                            s.request(requestPageSize);
                            System.out.println("onSubscribe:"+Thread.currentThread().getId());
                        }

                        @Override
                        public void onNext(Integer num) {
                            System.out.println("onNext:"+Thread.currentThread().getId()+",res:"+num);
                            try {
                                Thread.sleep(500);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            //模拟处理完一批资源之后，请求下一批资源
                            if(num<end && count.incrementAndGet() == requestPageSize){
                                s.request(requestPageSize);
                                count.set(0);
                            }
                        }

                        @Override
                        public void onError(Throwable t) {}

                        @Override
                        public void onComplete() {
                            System.out.println("消费者已经消费完了所有的数据:"+Thread.currentThread().getId());
                        }
                    });
                });
            }
            for (int i=0;i<5;i++){
                threads[i].start();
            }

            //响应式，Mono
            oneService.sayByeBye("张三").subscribe(System.out::println,null,()-> System.out.println("完成"));
            oneService.sayByeBye("李四").subscribe(System.out::println,Throwable::printStackTrace,()-> System.out.println("完成"));
            oneService.sayByeBye("王五").subscribe(r->{
                if(r==null){
                    System.out.println("空");
                }
            },null,()-> System.out.println("完成"));
            //同步形式
            System.out.println(oneService.sayHello());
            System.out.println(twoService.sayByeBye());
            System.out.println(twoService.getSum(1l,2l));

            //future异步形式
            oneService.getSum(3,3).whenComplete((sum,e)->{
               if(e!=null){
                   e.printStackTrace();
               }else {
                   System.out.println("oneService.getSum："+sum);
               }
            });

        }catch (Exception e){
            e.printStackTrace();
        }

    }

}
